package m3;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.Scanner;

/**
 * @description:
 * @author: LiuXingYu
 * @date: 2021/9/26 9:25
 */
public class Producer {

    public static void main(String[] args) throws MQClientException {


        // 创建事务消息生产者
        TransactionMQProducer p = new TransactionMQProducer("prod-group3");
        p.setNamesrvAddr("192.168.64.141:9876");

        p.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {

                // 模拟
                if (Math.random() < 1) {
                    System.out.println("网络中断, 无法向服务器提交或回滚");
                    return LocalTransactionState.ROLLBACK_MESSAGE; // 回滚消息, 撤回消息
                }

                System.out.println("执行本地业务, 参数: " + o);
                if (Math.random() < 0.5) {
                    System.out.println("本地事务执行成功");
                    return LocalTransactionState.COMMIT_MESSAGE; // 提交消息, 通知服务器, 消息可以投递
                }else {
                    System.out.println("本地事务执行失败");
                    return LocalTransactionState.ROLLBACK_MESSAGE; // 回滚消息, 撤回消息
                }
//                return LocalTransactionState.UNKNOW; // 未知, 当前方法中, 一般不会出现
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                System.out.println("rocketmq服务器正在回查事务状态");
                return LocalTransactionState.UNKNOW;
            }
        });

        p.start();
        while (true) {
            System.out.println("请输入: ");
            String s = new Scanner(System.in).nextLine();
            Message message = new Message("topic3", s.getBytes());
            TransactionSendResult result = p.sendMessageInTransaction(message, "业务执行参数");
            System.out.println("事务消息结果" + result);
        }



    }
}
